import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseImportJobData extends Thread {
    public Configuration config;
    public HTable table;
    public HBaseAdmin admin;

    public HBaseImportJobData() {
        config = HBaseConfiguration.create();
//      config.set("hbase.master", "master:60000");
//      config.set("hbase.zookeeper.quorum", "master");
        try {
            table = new HTable(config, Bytes.toBytes("job_java"));
            admin = new HBaseAdmin(config);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        String fileName = "/usr/local/bigdatacase/dataset/job_action.output";
        HBaseImportJobData importer = new HBaseImportJobData();
        importer.importLocalFileToHBase(fileName);
    }

    public void importLocalFileToHBase(String fileName) {
        long start = System.currentTimeMillis();
        BufferedReader br = null;
        try {
            br = new BufferedReader(new InputStreamReader(new FileInputStream(fileName), "UTF-8"));
            String line = null;
            int count = 0;
            while ((line = br.readLine()) != null) {
                count++;
                put(line);
                if (count % 10000 == 0)
                    System.out.println(count + " rows imported...");
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            try {
                table.flushCommits();
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("Total Time: " + (end - start) + " ms");
    }

    @SuppressWarnings("deprecation")
    public void put(String line) throws IOException {
        String[] arr = line.split(",", -1);
        String[] columns = {
            "id", "job_type", "job_title", "annual_salary", "company_type", "company_size", "city"
        };

        if (arr.length == 7) {
            Put put = new Put(Bytes.toBytes(arr[0])); // id as rowkey
            for (int i = 1; i < arr.length; i++) {
                put.add(Bytes.toBytes("f1"), Bytes.toBytes(columns[i]), Bytes.toBytes(arr[i]));
            }
            table.put(put);
        } else {
            System.err.println("Invalid line, skipped: " + line);
        }
    }

    public void get(String rowkey, String columnFamily, String column, int versions) throws IOException {
        long st = System.currentTimeMillis();
        Get get = new Get(Bytes.toBytes(rowkey));
        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        Scan scanner = new Scan(get);
        scanner.setMaxVersions(versions);
        ResultScanner rsScanner = table.getScanner(scanner);
        for (Result result : rsScanner) {
            final List<KeyValue> list = result.list();
            for (final KeyValue kv : list) {
                System.out.println(Bytes.toStringBinary(kv.getValue()) + "\t" + kv.getTimestamp());
            }
        }
        rsScanner.close();
        long en2 = System.currentTimeMillis();
        System.out.println("Total Time: " + (en2 - st) + " ms");
    }
}
